package org.example._02_send_way;

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;


public class Producer {
    public static void main(String[] args) throws Exception {
        //1 创建一个生产者对象, 并且指定一个生产者组(一个组里有多个生产者,多个生产者可在同一个组里面)
        DefaultMQProducer defaultMQProducer = new DefaultMQProducer("rocketmq-producer");
        //2 设置名字服务的地址(和 nameser 进行绑定)
        //如果出现 No route info of this topic 错误时,就是生产者找不到 rocketmq 了,可能是端口号写错了亦或者 mq 服务挂掉了(重启下)
        defaultMQProducer.setNamesrvAddr("127.0.0.1:9876");
        //3 启动生产者
        defaultMQProducer.start();

        for (int i = 0; i < 10; i++) {
            //4 创建一个消息对象,把数据封装到消息对象中
            Message message = new Message(
                    "02-send-way",//第一个参数  topic  主题
                    "tagA", //可选参数 标签
                    "hello,rocketmq".getBytes("utf-8")//第二个参数 byte[] 新消息内容
            );
            //5 发送消息(此处是异步发送消息)
            defaultMQProducer.send(message,new SendCallback(){

                /**
                 * 异步有回调,如果消息持久化没有问题,在回调的时候我们生产资源没有关闭那么 回调onSuccess方法
                 */
                @Override
                public void onSuccess(SendResult sendResult) {
                    System.out.println("发送成功");
                }

                /**
                 *  如果持久化有问题或者资源已经关闭就会调用onException方法
                 */
                @Override
                public void onException(Throwable throwable) {
                    System.out.println("发送失败");
                }
            });
        }

        Thread.sleep(10000);//由于异步,所以不卡下线程的话就直接关闭资源了导致回函时一直发送失败了
        //6 关闭连接(关闭资源)
        defaultMQProducer.shutdown();
    }
}